package com.byjc.aesb.consumer;

import com.alibaba.fastjson.JSON;
import com.byjc.aesb.config.RocketMQConfig;
import com.byjc.aesb.dto.RocketMQMessage;
import com.byjc.aesb.entity.RcmsBaseinfo;
import com.byjc.aesb.service.RcmsBaseinfoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;

/**
 * RocketMQ消费者
 *
 * @author byjc
 */
@Slf4j
@Component
public class RocketMQConsumer {

    @Autowired
    private RocketMQConfig rocketMQConfig;

    @Autowired
    private RcmsBaseinfoService rcmsBaseinfoService;

    private DefaultMQPushConsumer consumer;

    @PostConstruct
    public void init() throws MQClientException {
        log.info("初始化RocketMQ消费者...");
        RPCHook rpcHook = new AclClientRPCHook(new SessionCredentials(rocketMQConfig.getConsumer().getKey(), rocketMQConfig.getConsumer().getSecret()));
        consumer = new DefaultMQPushConsumer(null, rocketMQConfig.getConsumer().getGroup(), rpcHook, new AllocateMessageQueueAveragely(), true, null);
        consumer.setNamesrvAddr(rocketMQConfig.getNameServer());
        consumer.subscribe(rocketMQConfig.getConsumer().getTopic(), "*");
        consumer.setUseTLS(true);
        consumer.setPollNameServerInterval(10000);


        // 设置消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        String messageBody = new String(msg.getBody(), "UTF-8");
                        log.info("收到消息: {}", messageBody);

                        // 解析消息
                        RocketMQMessage rocketMQMessage = JSON.parseObject(messageBody, RocketMQMessage.class);

                        if (rocketMQMessage != null && rocketMQMessage.getData() != null) {
                            // 保存道口基本信息到数据库
                            boolean result = rcmsBaseinfoService.batchSave(rocketMQMessage.getData());
                            if (result) {
                                log.info("成功处理消息，bizId: {}, 保存记录数: {}",
                                        rocketMQMessage.getBizId(), rocketMQMessage.getData().size());
                            } else {
                                log.error("处理消息失败，bizId: {}", rocketMQMessage.getBizId());
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }
                        } else {
                            log.warn("消息格式不正确或数据为空，bizId: {}",
                                    rocketMQMessage != null ? rocketMQMessage.getBizId() : "unknown");
                        }

                    } catch (Exception e) {
                        log.error("处理消息异常", e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        log.info("RocketMQ消费者启动成功，Topic: {}, Group: {}",
                rocketMQConfig.getConsumer().getTopic(), rocketMQConfig.getConsumer().getGroup());
    }

    @PreDestroy
    public void destroy() {
        if (consumer != null) {
            consumer.shutdown();
            log.info("RocketMQ消费者已关闭");
        }
    }
} 